package cn.iocoder.yudao.framework.websocket.core.sender.rocketmq;

import cn.iocoder.yudao.framework.mq.rocketmq.api.consumer.handler.EnhanceMessageHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;

/**
 * {@link RocketMQWebSocketMessage} 广播消息的消费者，真正把消息发送出去
 *
 * @author 芋道源码
 */
@RocketMQMessageListener( // 重点：添加 @RocketMQMessageListener 注解，声明消费的 topic
        topic = "${yudao.websocket.sender-rocketmq.topic}",
        consumerGroup = "${yudao.websocket.sender-rocketmq.consumer-group}",
        messageModel = MessageModel.BROADCASTING // 设置为广播模式，保证每个实例都能收到消息
)
@RequiredArgsConstructor
@Slf4j
public class RocketMQWebSocketMessageConsumer extends EnhanceMessageHandler<RocketMQWebSocketMessage> implements RocketMQListener<RocketMQWebSocketMessage> {

    private final RocketMQWebSocketMessageSender rocketMQWebSocketMessageSender;


    /**
     * 消息处理
     *
     * @param message 待处理消息
     * @throws Exception 消费异常
     */
    @Override
    protected void handleMessage(RocketMQWebSocketMessage message)  {
        try {
            log.info("消费消息：{}", message);
            rocketMQWebSocketMessageSender.send(message.getSessionId(),
                    message.getUserType(), message.getUserId(),
                    message.getMessageType(), message.getMessageContent());
        } catch (Exception e) {
            log.error("MQ消费失败",e);
            throw new RuntimeException(e.getMessage());
        }
    }

    @Override
    public void onMessage(RocketMQWebSocketMessage message) {
        try {
            dispatchMessage(message);
        } catch (Exception e) {
            log.error("MQ消费失败", e);
            throw new RuntimeException(e.getMessage());
        }
    }
}
